package pku;

import java.io.*;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

public class MessageStoreArgs {
    static final MessageStoreArgs store = new MessageStoreArgs();
    File file = new File("./data/a.txt");
    OutputStream out;
    InputStream in;
    //给每个consumer对应一个流
    HashMap<String, InputStream> inMap = new HashMap<String, InputStream>();

    // 同步执行push方法，防止多个线程同时访问互斥资源
    public synchronized void push(ByteMessage msg) {
        if (msg == null) {
            return;
        }
        //获取topic
        String topic = msg.headers().getString(MessageHeader.TOPIC);

        try {
            if (out == null) {
                out = new FileOutputStream(file, true);
            }

//            HashMap<String, Object> hm = msg.headers().getMap();
            HashMap<String, Object> hm = new HashMap<>();
            int len = msg.getBody().length + topic.getBytes().length + 3;
            for (String k : hm.keySet()) {
                if (MessageHeader.TOPIC.equals(k)) {
                    continue;
                }
                int lenHeader = k.getBytes().length + hm.get(k).toString().getBytes().length + 3;
                len += lenHeader;
            }

            byte[] byteTotal = new byte[len];

            byteTotal[0] = (byte) (len - 1);

            byteTotal[1] = (byte) topic.getBytes().length;
            byte[] tb = topic.getBytes();
            System.arraycopy(tb, 0, byteTotal, 2, tb.length);

            byteTotal[tb.length + 2] = (byte) msg.getBody().length;
            byte[] bo = msg.getBody();
            System.arraycopy(bo, 0, byteTotal, tb.length + 3, bo.length);

            int length = tb.length + 2 + bo.length;
//            System.out.println("length="+length);
            for (String key : hm.keySet()) {
                if (MessageHeader.TOPIC.equals(key)) {
                    continue;
                }

                Object obj = hm.get(key);

                byteTotal[++length] = (byte) key.getBytes().length;

                byte[] kb = key.getBytes();
                System.arraycopy(kb, 0, byteTotal, ++length, kb.length);
                length = length + kb.length - 1;

                int type = type(obj);
                byteTotal[++length] = (byte) type;

                byteTotal[++length] = (byte) obj.toString().getBytes().length;

                byte[] hv = obj.toString().getBytes();
                System.arraycopy(hv, 0, byteTotal, ++length, hv.length);
                length = length + hv.length - 1;
            }
            out.write(byteTotal);

        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    public synchronized ByteMessage pull(String queue, List<String> topics) {
        try {
            if (!inMap.containsKey(queue))
                inMap.put(queue, new FileInputStream(file));
            //每个queue都有一个InputStream
            in = inMap.get(queue);
            if (in.available() == 0) {
                return null;
            }

            //每次循环读一个message的数据量
            String ss = "";
            do {
                byte lenTotal = (byte) in.read();
                //读到文件尾了，则lenTotal为-1
                if (lenTotal == -1)
                    return null;
                byte[] byteTotal = new byte[lenTotal];
                in.read(byteTotal);

                byte lenTopic = byteTotal[0];
                byte[] topic = new byte[lenTopic];
                System.arraycopy(byteTotal, 1, topic, 0, lenTopic);
                ss = new String(topic);

                if (topics.contains(ss)) {
                    byte lenBody = byteTotal[lenTopic + 1];
                    byte[] body = new byte[lenBody];
                    System.arraycopy(byteTotal, lenTopic + 2, body, 0, lenBody);

                    ByteMessage msg = new DefaultMessage(body);
                    msg.putHeaders(MessageHeader.TOPIC, new String(topic));

                    int lenTB = lenTopic + lenBody + 1;
                    while (lenTB < lenTotal - 1) {
                        byte lenHeader = byteTotal[++lenTB];
                        byte[] headerKey = new byte[lenHeader];
                        System.arraycopy(byteTotal, ++lenTB, headerKey, 0, lenHeader);
                        String key = new String(headerKey);

                        lenTB = lenTB + lenHeader - 1;
                        byte headerType = byteTotal[++lenTB];

                        byte lenValue = byteTotal[++lenTB];
                        byte[] headerValue = new byte[lenValue];
                        System.arraycopy(byteTotal, ++lenTB, headerValue, 0, lenValue);
                        lenTB = lenTB + lenValue - 1;

                        if (headerType == 3) {
                            String s = new String(headerValue);
                            msg.headers().put(key, s);
                        } else if (headerType == 4) {
                            int i = Integer.valueOf(new String(headerValue));
                            msg.headers().put(key, i);
                        } else if (headerType == 1) {
                            long l = Long.valueOf(new String(headerValue));
                            msg.headers().put(key, l);
                        } else if (headerType == 2) {
                            double d = Double.valueOf(new String(headerValue));
                            msg.headers().put(key, d);
                        }
                    }
                    return msg;
                }

            } while (!topics.contains(ss));

        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;

    }

    private int type(Object obj) {

        if (obj instanceof Integer) {
            return 4;
        } else if (obj instanceof Long) {
            return 1;
        } else if (obj instanceof Double) {
            return 2;
        } else if (obj instanceof String) {
            return 3;
        }
        return 3;
    }
}
